package sql_wc;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.db.DBConfiguration;
import org.apache.hadoop.mapreduce.lib.db.DBInputFormat;
import org.apache.hadoop.mapreduce.lib.db.DBOutputFormat;
import org.apache.log4j.BasicConfigurator;

import java.io.IOException;

public class DriTest {
    public static void main(String[] args) {
//        BasicConfigurator.configure();
//        Configuration conf = new Configuration();
//        DBConfiguration.configureDB(conf, "com.mysql.jdbc.Driver", "jdbc:mysql://localhost:3306/data", "root", "200028");
//        Job job = Job.getInstance(conf);
//        job.setJarByClass(NewOutputFormatDriver.class);
//
//        job.setMapperClass(NewOutputFormatMapper.class);
//        job.setMapOutputKeyClass(Text.class);
//        job.setMapOutputValueClass(IntWritable.class);
//
//        job.setReducerClass(NewOutputFormatReduce.class);
//        job.setOutputKeyClass(SendTable.class);
//        job.setOutputValueClass(NullWritable.class);
//
//        String[] f1 = {"word"};
//        String[] f2 = {"word", "count"};
//
//        DBInputFormat.setInput(job, ReceiveTable.class, "WC_R", null, null, f1);
//        DBOutputFormat.setOutput(job, "WC_S", f2);
//        System.exit(job.waitForCompletion(true) ? 0 : 1);
        try {
            BasicConfigurator.configure();
            Configuration conf = new Configuration();
            conf.set("fs.defaultFS","hdfs://d:9000");

            DBConfiguration.configureDB(conf, "com.mysql.jdbc.Driver", "jdbc:mysql://192.168.0.155:3306/data", "root", "123456");
            Job job = Job.getInstance(conf,"mr");
            job.setJarByClass(DriTest.class);

            job.setMapperClass(MapTest.class);
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(IntWritable.class);

            job.setReducerClass(RedTest.class);
            job.setOutputKeyClass(SendTable.class);
            job.setOutputValueClass(NullWritable.class);

            String[] f1 = {"word"};
            String[] f2 = {"word", "count"};
            DBInputFormat.setInput(job, ReceiveTable.class, "wcr", null, "word", f1);
            DBOutputFormat.setOutput(job,"wcs",f2);
            System.exit(job.waitForCompletion(true) ? 0 : 1);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }
    }
}
